package com.ssm.mq.config;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.TransactionManager;

import javax.annotation.PostConstruct;

@Slf4j
@RequiredArgsConstructor
@Configuration(proxyBeanMethods = false)
public class AmqpConfig {
    final RabbitTemplate rabbitTemplate; // final AmqpTemplate amqpTemplate;
    public static final String ROUTING_KEY = "";
    public static final String QUEUE_NAME = "queueName";
    public static final String EXCHANGE_NAME = "exchageName";
    public static final String QUEUE_NAME_ALTERNATE = "queueNameAlternate";
    public static final String EXCHANGE_NAME_ALTERNATE = "exchageNameAlternate";
    public static final String QUEUE_NAME_TTL = "queueNameTTL";
    public static final String QUEUE_NAME_DLX = "queueNameDLX";
    public static final String EXCHANGE_NAME_DLX = "exchageNameDLX";
    public static final String QUEUE_NAME_ANONYMOUS = "queueNameAnonymous";

    @Bean
    Exchange exchange() {
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME)
            .alternate(EXCHANGE_NAME_ALTERNATE) // 备份交换机，接收没有路由到 QUEUE 的消息。优先级高于 ReturnsCallback
            .build();
    }

    @Bean
    Queue queue() {
        return QueueBuilder.durable(QUEUE_NAME)
            .maxPriority(10) // 优先级 0-10，最大 255
            .build();
    }

    @Bean
    Binding binding(Queue queue, Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY).noargs();
    }

    // ============================================================================================================

    @Bean
    FanoutExchange alternateExchange() {
        return ExchangeBuilder.fanoutExchange(EXCHANGE_NAME_ALTERNATE).build();
    }

    @Bean
    Queue alternateQueue() {
        return QueueBuilder.durable(QUEUE_NAME_ALTERNATE).build();
    }

    @Bean
    Binding alternateBinding(Queue alternateQueue, FanoutExchange alternateExchange) {
        return BindingBuilder.bind(alternateQueue).to(alternateExchange);
    }

    // ============================================================================================================

    @Bean
    Queue anonymousQueue() { // 一个匿名的、不耐用的、独占的、自动删除的队列
        return new AnonymousQueue(() -> QUEUE_NAME_ANONYMOUS);
    }

    // ============================================================================================================

    @Bean
    Queue ttlQueue() { // https://www.rabbitmq.com/ttl.html
        return QueueBuilder.nonDurable(QUEUE_NAME_TTL)
            .lazy() // 将消息保存在磁盘上，即使队列被设为 nonDurable(不持久化)，消息依然会被写入到磁盘
            .ttl(6000) // 队列中消息 6 秒后自动删除
            .maxLength(3) // 队列中最多只有 3 个消息，https://www.rabbitmq.com/maxlength.html
            .deadLetterExchange(EXCHANGE_NAME_DLX) // 当消息在队列中变成死信(dead message)后，它会被 publish 到另一个 Exchange
            .deadLetterRoutingKey(ROUTING_KEY)
            .build();
    }

    @Bean
    Exchange dlxExchange() { // 死信交换机 https://www.rabbitmq.com/dlx.html
        // 死信是指有这三个特点的消息：消息被拒绝，且没有重新入队(投递)、消息过期、消息队列满了
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME_DLX).build();
    }

    @Bean
    Queue dlxQueue() { // 死信队列：DLX，dead-letter-exchange。可以用来做延迟队列
        return QueueBuilder.durable(QUEUE_NAME_DLX).build();
    }

    @Bean
    Binding dlxBinding(Queue dlxQueue, Exchange dlxExchange) {
        return BindingBuilder.bind(dlxQueue).to(dlxExchange).with(ROUTING_KEY).noargs();
    }

    // ========================================================================================================

    @PostConstruct
    public void initRabbitTemplate() {
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("发送消息到 EXCHANGE 成功");
            } else {
                log.error("发送消息到 EXCHANGE 失败：" + cause);
            }
        });
        rabbitTemplate.setReturnsCallback(returned -> log.error("EXCHANGE 路由消息到 QUEUE 失败：" + returned.getMessage().toString()));
    }

    /**
     * 启用 rabbitmq 事务，生产中不建议使用事务模式，性能比较低，尽量使用确认模式。启用事务模式，不能开确认回调
     */
    // @Bean
    TransactionManager transactionManager(CachingConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }
}